package Log

import (
	"context"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"log"
	"time"
)

/*
type asyncRocketMQWriter struct {
	producer rocketmq.Producer
	queue    chan []byte
	wg       sync.WaitGroup
	quit     chan struct{}
	isClosed bool
}

func newAsyncRocketMQWriter() *asyncRocketMQWriter {
	p, err := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.3.242:9876"})),
		producer.WithRetry(2),
		producer.WithSendMsgTimeout(10*time.Second), // 增加超时时间为 10 秒
	)
	if err != nil {
		log.Fatal(err)
	}
	err = p.Start()
	if err != nil {
		log.Fatal(err)
	}

	w := &asyncRocketMQWriter{
		producer: p,
		queue:    make(chan []byte, 10000000),
		quit:     make(chan struct{}),
	}
	w.wg.Add(1)
	go w.run()
	return w
}

func (w *asyncRocketMQWriter) Write(p []byte) (n int, err error) {
	if w.isClosed {
		return 0, io.ErrClosedPipe
	}
	w.queue <- p
	return len(p), nil
}

func (w *asyncRocketMQWriter) run() {
	defer w.wg.Done()
	for {
		select {
		case data := <-w.queue:
			go func(d []byte) {
				_, err := w.producer.SendSync(context.Background(), primitive.NewMessage("AdminLogs", d))
				if err != nil {
					log.Println("发送到 RocketMQ 错误：", err)
				}
			}(data)
		case <-w.quit:
			return
		}
	}
}

func (w *asyncRocketMQWriter) Sync() error {
	w.isClosed = true
	for len(w.queue) > 0 {
		time.Sleep(time.Millisecond * 100)
	}
	close(w.quit)
	w.wg.Wait()
	return nil
}

func (w *asyncRocketMQWriter) Close() error {
	err := w.Sync()
	if err != nil {
		return err
	}
	return w.producer.Shutdown()
}

func InitLoggerWithRocketMQ() *zap.Logger {
	logger, err := zap.NewProduction()
	if err != nil {
		log.Fatal(err)
	}

	asyncWriter := newAsyncRocketMQWriter()
	logger = logger.WithOptions(zap.WrapCore(func(core zapcore.Core) zapcore.Core {
		return zapcore.NewCore(zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), asyncWriter, zap.InfoLevel)
	}))

	return logger
}
*/
//同步

type syncRocketMQWriter struct {
	producer rocketmq.Producer
}

func newSyncRocketMQWriter() *syncRocketMQWriter {
	p, err := rocketmq.NewProducer(
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.3.242:9876"})),
		producer.WithRetry(2),
		producer.WithSendMsgTimeout(10*time.Second),
	)
	if err != nil {
		log.Fatal(err)
	}
	err = p.Start()
	if err != nil {
		log.Fatal(err)
	}

	return &syncRocketMQWriter{
		producer: p,
	}
}

func (w *syncRocketMQWriter) Write(p []byte) (n int, err error) {
	_, err = w.producer.SendSync(context.Background(), primitive.NewMessage("AdminLogs", p))
	if err != nil {
		return 0, err
	}
	return len(p), nil
}

func (w *syncRocketMQWriter) Sync() error {
	// 这里可以根据需要添加一些同步操作，比如确保所有消息都已发送成功等
	return nil
}

func (w *syncRocketMQWriter) Close() error {
	err := w.producer.Shutdown()
	if err != nil {
		return err
	}
	return nil
}

func InitLoggerWithRocketMQ() *zap.Logger {
	logger, err := zap.NewProduction()
	if err != nil {
		log.Fatal(err)
	}

	syncWriter := newSyncRocketMQWriter()
	logger = logger.WithOptions(zap.WrapCore(func(core zapcore.Core) zapcore.Core {
		return zapcore.NewCore(zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), syncWriter, zap.InfoLevel)
	}))

	return logger
}
